/**     
 * @{#} DefaultConnectPoolImpl.java Create on 2011-7-22 上午09:59:05     
 *     
 * Copyright (c) 2011 by xiaofeng.zhouxf     
 */
package com.sefer.memcached.connectpool;

import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

import com.sefer.memcached.arithmetic.ArithmeticFactory;
import com.sefer.memcached.arithmetic.HashArithmetic;
import com.sefer.memcached.connect.ConnectResources;
import com.sefer.memcached.connect.Connector;
import com.sefer.memcached.connect.DefaultConnectorImpl;
import com.sefer.memcached.connect.Server;
import com.sefer.memcached.constants.HashPolicy;
import com.sefer.memcached.exception.ClientException;
import com.sefer.memcached.utils.Assert;
import com.sefer.memcached.utils.LogUtils;

/**
 * This class is default pool implement. It will use arithmetic to calculate
 * witch connector is suit for the key.
 * 
 * @author <a href="mailto:xiaofeng.zhouxf83@gmail.com">xiaofeng.zhouxf</a>
 * @version 1.0
 */
public class DefaultConnectPoolImpl implements ConnectPool {
	private static final Logger log = Logger
			.getLogger(DefaultConnectPoolImpl.class.getName());
	/**
	 * Hash policy for arithmetic.The default arithmetic is mod hash.
	 */
	private HashPolicy hashPolicy = HashPolicy.MOD_HASH;
	private String lifeCycleStats;
	private ConnectResources resources;
	// every server has number's of instance
	private ConnectorMetainor[] connectorMetainor;
	// The default connect count for each server.
	private int DEFAULT_CONNECT_LENGTH = 10;
	private ScheduledExecutorService scheduledThreadPool;
	private ScheduledFuture scheduleAtFixedRate;

	@Override
	public boolean start() throws ClientException {
		Assert.assertTrue(validateParams(),
				"pool start error for params is error.");
		Server[] servers = resources.getServers();
		connectorMetainor = new ConnectorMetainor[servers.length];
		int i = 0;

		for (Server server : servers) {
			Connector[] connectorArray = new Connector[DEFAULT_CONNECT_LENGTH];
			for (int j = 0; j < DEFAULT_CONNECT_LENGTH; j++) {
				connectorArray[j] = new DefaultConnectorImpl();
				connectorArray[j].setServer(server);
				connectorArray[j].setState(getState());
				connectorArray[j].start();
			}
			connectorMetainor[i] = new ConnectorMetainor();
			connectorMetainor[i].setConnectors(connectorArray);
			i++;
		}

		// Start a thread for metian the socket connect's count
		scheduledThreadPool = Executors.newScheduledThreadPool(1);
		scheduleAtFixedRate = scheduledThreadPool.scheduleAtFixedRate(
				new Runnable() {
					@Override
					public void run() {
						LogUtils.info(log, "schedule is running");
						for (int i = 0; i < connectorMetainor.length; i++) {
							Connector[] connectors = connectorMetainor[i]
									.getConnectors();
							for (Connector connector : connectors) {
								if (connector.getSocketWrap().getSocket()
										.isClosed()) {
									connector.start();
								}
							}
						}
					}
				}, 1, 2, TimeUnit.SECONDS);

		Runtime.getRuntime().addShutdownHook(new Thread() {
			public void run() {
				LogUtils.info(log, "The connect pool had been destoried.");
				for (int i = 0; i < connectorMetainor.length; i++) {
					Connector[] connectors = connectorMetainor[i]
							.getConnectors();
					for (Connector connector : connectors) {
						if (connector.getSocketWrap().getSocket().isConnected()) {
							try {
								connector.getSocketWrap().getSocket().close();
							} catch (IOException e) {
								LogUtils.error(log, e.getMessage());
							}
						}
					}
				}
			}
		});
		return true;
	}

	@Override
	public boolean stop() throws ClientException {
		scheduleAtFixedRate.cancel(true);
		scheduledThreadPool.shutdownNow();
		return true;
	}

	@Override
	public String getState() {
		return lifeCycleStats;
	}

	@Override
	public void setState(String state) {
		this.lifeCycleStats = state;
	}

	@Override
	public Connector getConnector(Object key) {

		HashArithmetic arithmetic = ArithmeticFactory.getArithmetic(hashPolicy);
		int calculate = arithmetic
				.calculate(key, resources.getServers().length);
		if (calculate >= 0 && calculate < connectorMetainor.length) {
			return connectorMetainor[calculate].getOneConnectors();
		}
		return null;
	}

	@Override
	public boolean initPool(ConnectResources resources) {
		this.resources = resources;
		return true;
	}

	@Override
	public void setArithmetic(HashPolicy policy) {
		this.hashPolicy = policy;
	}

	// =============
	boolean validateParams() throws ClientException {
		return true;
	}

	// ===============

	public ConnectResources getResources() {
		return resources;
	}

	public void setResources(ConnectResources resources) {
		this.resources = resources;
	}

	class ConnectorMetainor {
		Connector[] connectors;
		// the index for this metainor
		volatile int currentIndex;

		public synchronized Connector getOneConnectors() {
			if (currentIndex == connectors.length && currentIndex != 0) {
				currentIndex = 0;
			}
			Connector connector = null;
			while (true) {
				connector = connectors[currentIndex];
				currentIndex++;
				if (connector.getSocketWrap().getSocket().isConnected()) {
					break;
				}
			}
			return connector;
		}

		public Connector[] getConnectors() {
			return connectors;
		}

		public void setConnectors(Connector[] connectors) {
			this.connectors = connectors;
		}

		public void reset() {
			this.currentIndex = 0;
		}

	}
}
